package com.sh.data.engine.domain.integration.documentCapture.service.impl;

import cn.hutool.core.text.csv.CsvData;
import cn.hutool.core.text.csv.CsvReader;
import cn.hutool.core.text.csv.CsvRow;
import cn.hutool.core.text.csv.CsvUtil;
import cn.hutool.core.util.CharsetUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.sh.data.engine.common.constants.Constants;
import com.sh.data.engine.common.enumDefinition.DSType;
import com.sh.data.engine.common.enumDefinition.OfflineSyncTaskStatus;
import com.sh.data.engine.common.enumDefinition.TaskType;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.TableNameUtil;
import com.sh.data.engine.domain.authority.service.DataAuthService;
import com.sh.data.engine.domain.base.model.PageResult;
import com.sh.data.engine.domain.base.model.UserContext;
import com.sh.data.engine.domain.base.model.datax.DataX;
import com.sh.data.engine.domain.base.model.entity.TbTaskFieldMapEntity;
import com.sh.data.engine.domain.base.model.enums.RightEnum;
import com.sh.data.engine.domain.base.service.LogService;
import com.sh.data.engine.domain.common.service.TableManager;
import com.sh.data.engine.domain.dataasset.model.domain.TableInfoBizDomain;
import com.sh.data.engine.domain.dataasset.model.param.QueryTableParam;
import com.sh.data.engine.domain.dataasset.service.TableService;
import com.sh.data.engine.domain.integration.datasource.service.DataSourceService;
import com.sh.data.engine.domain.integration.documentCapture.model.domain.DocumentCaptureDomain;
import com.sh.data.engine.domain.integration.documentCapture.model.mapper.DocumentCaptureDomainMapper;
import com.sh.data.engine.domain.integration.documentCapture.model.param.SaveTaskWithFromTextParam;
import com.sh.data.engine.domain.integration.documentCapture.service.DocumentCaptureService;
import com.sh.data.engine.domain.integration.offline.model.domain.OfflineSyncDomain;
import com.sh.data.engine.domain.integration.offline.model.mapper.FieldMappingMapper;
import com.sh.data.engine.domain.integration.offline.service.DataXService;
import com.sh.data.engine.domain.shims.db.BaseDbManager;
import com.sh.data.engine.domain.shims.db.model.TableInfoDomain;
import com.sh.data.engine.domain.util.ConvertUtil;
import com.sh.data.engine.domain.util.FileDownloadUtil;
import com.sh.data.engine.domain.workspace.index.service.ProjectService;
import com.sh.data.engine.domain.workspace.manager.model.domain.UserDomain;
import com.sh.data.engine.domain.workspace.manager.service.UserService;
import com.sh.data.engine.domain.workspace.manager.service.impl.StorageServiceImpl;
import com.sh.data.engine.infrastructure.config.DataXConfig;
import com.sh.data.engine.infrastructure.config.FileStorageConfiguration;
import com.sh.data.engine.repository.dao.integration.datasource.entity.DataSourceEntity;
import com.sh.data.engine.repository.dao.integration.documentCapture.DocumentCaptureMapper;
import com.sh.data.engine.repository.dao.integration.documentCapture.entity.DocumentCaptureEntity;
import com.sh.data.engine.repository.dao.normalization.standard.entity.DictionaryEntity;
import com.sh.data.engine.repository.dao.workspace.manager.entity.StorageEntity;
import jakarta.servlet.http.HttpServletResponse;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.poi.hssf.usermodel.HSSFWorkbook;
import org.apache.poi.openxml4j.opc.OPCPackage;
import org.apache.poi.ss.usermodel.*;
import org.apache.poi.util.IOUtils;
import org.apache.poi.xssf.eventusermodel.XSSFReader;
import org.apache.poi.xssf.eventusermodel.XSSFSheetXMLHandler;
import org.apache.poi.xssf.model.SharedStrings;
import org.apache.poi.xssf.model.StylesTable;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;
import org.xml.sax.InputSource;
import org.xml.sax.SAXException;
import org.xml.sax.XMLReader;
import org.xml.sax.helpers.XMLReaderFactory;

import java.io.*;
import java.net.HttpURLConnection;
import java.net.URL;
import java.net.URLEncoder;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author: mengzheng.mhc
 * @date: 2024/6/26 9:38
 */
@Service
@Slf4j
public class DocumentCaptureServiceImpl
        extends ServiceImpl<DocumentCaptureMapper, DocumentCaptureEntity>
        implements DocumentCaptureService {

    @Autowired
    private TableService tableService;

    @Autowired
    private ProjectService projectService;

    @Autowired
    private StorageServiceImpl storageServiceImpl;

    @Autowired
    private DataXConfig dataXConfig;

    @Autowired
    private FileStorageConfiguration fileStorageConfiguration;

    @Autowired
    private DataXService dataXService;

    @Autowired
    private LogService logService;

    @Autowired
    private DataSourceService dataSourceService;

    @Autowired
    private UserService userService;

    @Autowired
    private TableManager tableManager;

    @Autowired
    private DataAuthService dataAuthService;

    @Autowired
    private FieldMappingMapper fieldMappingMapper;

    @Autowired
    private DocumentCaptureDomainMapper documentCaptureDomainMapper;


    @Override
    public List<Map<String, String>> fileParsingCSV(MultipartFile file, String encodeFormat) {
        InputStreamReader inputStreamReader = encodeFile(file, encodeFormat);
        //csv读取
        CsvReader reader = CsvUtil.getReader();
        //从文件中读取CSV数据
        CsvData csvData = reader.read(inputStreamReader);
        List<CsvRow> rows = csvData.getRows();
        //遍历行
        List<Map<String, String>> dataList = new ArrayList<>();
        for (int i = 0, rowsSize = rows.size(); i < Math.min(rowsSize, 11); i++) {
            Map<String, String> data = new LinkedHashMap<>();
            CsvRow csvRow = rows.get(i);
            //getRawList返回一个List列表，列表的每一项为CSV中的一个单元格（既逗号分隔部分）
            List<String> rawList = csvRow.getRawList();
            for (int j = 0; j < rawList.size(); j++) {
                data.put(i + "_" + j, rawList.get(j));
            }
            dataList.add(data);
        }
        return dataList;
    }

    /**
     * 文件编码格式转换
     *
     * @param file         输入文件
     * @param encodeFormat 文件编码格式
     * @return Reader
     */
    @NotNull
    private static InputStreamReader encodeFile(MultipartFile file, String encodeFormat) {
        InputStreamReader inputStreamReader;
        try {
            if ("GBK".equals(encodeFormat)) {
                inputStreamReader = new InputStreamReader(file.getInputStream(), CharsetUtil.CHARSET_GBK);
            } else if ("UTF-8".equals(encodeFormat)) {
                inputStreamReader = new InputStreamReader(file.getInputStream(), CharsetUtil.CHARSET_UTF_8);
            } else {
                throw new BusinessException("请传入正确的编码格式: GBK or UTF-8");
            }
        } catch (IOException e) {
            throw new BusinessException("解析Csv文件出错" + e);
        }
        return inputStreamReader;
    }

    @Override
    public List<TableInfoBizDomain> getTableList(Long projectId) {
        QueryTableParam queryTableParam = new QueryTableParam();
        List<TableInfoBizDomain> tableList = tableService.getTableList(projectId, queryTableParam);
        // 过滤掉外表
        List<TableInfoBizDomain> result =
                tableList.stream()
                        .filter(tl -> tl.getTblType() != TableInfoDomain.TblType.EXTERNAL_TABLE)
                        .collect(Collectors.toList());
        Iterator<TableInfoBizDomain> iterator = result.iterator();
        String userId = UserContext.getUserId();
        while (iterator.hasNext()) {
            TableInfoDomain next = iterator.next();
            Long tblId = next.getTblId();
            if (tblId == null) {
                continue;
            }
            String tblAuth = dataAuthService.getTblAuth(projectId, userId, tblId);
            if (StringUtils.isBlank(tblAuth)) {
                continue;
            }
            // 目标：无写权限，过滤掉
            if (!RightEnum.RW.getCode().equals(tblAuth)) {
                iterator.remove();
            }
        }
        return result;
    }

    @Override
    public Boolean saveTaskWithFromText(SaveTaskWithFromTextParam saveTaskWithTextParam) {

        // 定义entity存储业务数据
        DocumentCaptureEntity documentCaptureEntity = new DocumentCaptureEntity();
        documentCaptureEntity.setOrgFileName(saveTaskWithTextParam.getFileName());
        // 存入下载地址
        documentCaptureEntity.setOrgFilePath(saveTaskWithTextParam.getFileDownloadPath());
        documentCaptureEntity.setTargetTblId(saveTaskWithTextParam.getTblId());
        documentCaptureEntity.setTargetTblName(saveTaskWithTextParam.getTableName());
        documentCaptureEntity.setState(2);
        documentCaptureEntity.setUploadNum(0);
        documentCaptureEntity.setProjectId(saveTaskWithTextParam.getProjectId());
        documentCaptureEntity.setDataReplaceRuleId(saveTaskWithTextParam.getDataReplaceRuleId());
        documentCaptureEntity.setUpdaterId(UserContext.getUserId());
        documentCaptureEntity.setUpdateTime(new Date());
        this.saveOrUpdate(documentCaptureEntity);

        StorageEntity storageEntity =
                storageServiceImpl.getStorageInfoByProjectId(saveTaskWithTextParam.getProjectId());
        // 目标存储类型
        String storageType = storageEntity.getStorageType();

        OfflineSyncDomain offlineSyncDomain = new OfflineSyncDomain();
        offlineSyncDomain.setProjectId(saveTaskWithTextParam.getProjectId());

        // 获取dsId

        DataSourceEntity systemDataSourceDetail = dataSourceService.getSystemDataSourceDetail(
                saveTaskWithTextParam.getProjectId());
        if (systemDataSourceDetail == null) {
            throw new BusinessException("获取当前项目的数据源失败");
        }
        offlineSyncDomain.setTargetDsId(systemDataSourceDetail.getId());
        offlineSyncDomain.setTargetTable(
                TableNameUtil.getTblNameShort(saveTaskWithTextParam.getTableName()));
        offlineSyncDomain.setTaskName("文件采集");
        // 任务类型
        offlineSyncDomain.setTaskType(TaskType.text.getTaskType());
        offlineSyncDomain.setDataReplaceRuleId(saveTaskWithTextParam.getDataReplaceRuleId());
        // 源数据源类型
        offlineSyncDomain.setSourceDsType(DSType.Text.name());
        // 目标数据源类型
        offlineSyncDomain.setTargetDsType(storageType);
        offlineSyncDomain.setTaskStatus(OfflineSyncTaskStatus.torelease.getTaskStatus());
        offlineSyncDomain.setCreateTime(new Date());
        offlineSyncDomain.setUpdateTime(new Date());
        offlineSyncDomain.setSkipCount(saveTaskWithTextParam.getSkipCount());
        // csv文件的编码格式
        offlineSyncDomain.setEncodeFormat(saveTaskWithTextParam.getEncodeFormat());
        if (storageType.equals(DSType.PostgreSQL.name()) || storageType.equals(DSType.Sap.name())
                || storageType.equalsIgnoreCase(DSType.KINGBASE8.name())) {
            // 如果底层数据源是pg或者是sap，要传入schema
            // String schema = storageEntity.getRmdbs().getSchema();
            String schema = TableNameUtil.getTblSchema(saveTaskWithTextParam.getTableName());
            offlineSyncDomain.setSchema(schema);
        } else if (storageType.equals(DSType.DM.name())) {
            offlineSyncDomain.setSchema(systemDataSourceDetail.getRdbmsConfig().getDbName());
        }

        // 目标是RDBMS/hive走 datax

        // 获取字段映射信息
        List<TbTaskFieldMapEntity> tbTaskFieldMapEntities =
                saveTaskWithTextParam.getTbTaskFieldMapEntities();
        // 需要把字段关系映射中的字段加个双引号 因为pg会把所有的大写改成小写
        // 如果是Sap一样要加双引号  会报错 invalid column name: COLUMN0: line 1 col 8 (at pos 7)，Sap 会把字段全部变大写？裂开了
        if (DSType.PostgreSQL.name().equals(storageType) || storageType.equals(DSType.Sap.name())
                || storageType.equalsIgnoreCase(DSType.KINGBASE8.name())) {
            for (TbTaskFieldMapEntity tbTaskFieldMapEntity : tbTaskFieldMapEntities) {
                String targetField = tbTaskFieldMapEntity.getTargetField();
                tbTaskFieldMapEntity.setTargetField("\"" + targetField + "\"");
            }
        }

        List<OfflineSyncDomain.FieldMapping> fieldMappings = fieldMappingMapper.map(tbTaskFieldMapEntities);

        // 获取文件地址
        String orgFilePath = saveTaskWithTextParam.getFilePath();

        offlineSyncDomain.setFieldMappings(fieldMappings);
        offlineSyncDomain.setFilePath(orgFilePath);

        // 构建datax任务并使用datax进行离线任务同步
        DataX dataX = dataXService.buildDataXTextJob(offlineSyncDomain);

        String jsonScript = dataXService.setVarsText(dataX);

        String dataxHome = dataXConfig.getHome();

        String dataxJsonPath =
                String.join(
                        File.separator,
                        dataxHome,
                        "job",
                        "documentCapture",
                        documentCaptureEntity.getId() + ".json");

        String logDir = fileStorageConfiguration.getDir();

        DocumentCaptureRunnable task =
                new DocumentCaptureRunnable(jsonScript, dataxJsonPath, documentCaptureEntity, logDir);

        Thread thread = new Thread(task);

        thread.start();
//        LogContext.put(LogKey.NAME.getKey(), saveTaskWithTextParam.getTableName());

        return true;
    }

    @Override
    public PageResult<DocumentCaptureDomain> getCaptureList(
            String tableName, Integer pageNum, Integer pageSize, Long projectId) {

        Page<DocumentCaptureEntity> page =
                this.page(
                        new Page<>(pageNum, pageSize),
                        new LambdaQueryWrapper<DocumentCaptureEntity>()
                                .eq(DocumentCaptureEntity::getProjectId, projectId)
                                .like(
                                        StringUtils.isNotBlank(tableName),
                                        DocumentCaptureEntity::getTargetTblName,
                                        tableName)
                                .orderByDesc(DocumentCaptureEntity::getCreateTime));
        PageResult<DocumentCaptureEntity> documentCaptureEntityPageResult =
                new PageResult<>(page.getTotal(), page.getRecords(), pageNum, pageSize);

        PageResult<DocumentCaptureDomain> documentCaptureDomainPageResult =
                ConvertUtil.convertPageInfo(documentCaptureEntityPageResult, DocumentCaptureDomain.class);
        // 下面去拼装创建人name
        for (DocumentCaptureDomain documentCaptureDomain :
                documentCaptureDomainPageResult.getResult()) {

            UserDomain userById = userService.getUserById(documentCaptureDomain.getUpdaterId());
            if (userById != null) {
                documentCaptureDomain.setCreatorName(userById.getUserName());
            }
        }

        return documentCaptureDomainPageResult;
    }

    @Override
    public List<String> getLog(Long docCapId) {
        String logDir = fileStorageConfiguration.getDir();
        String subTaskJobLogPath =
                String.format("%s/%s/log/%s.log", logDir, Constants.TEXT_EXPORT_LOG, docCapId);
        try {
            List<String> read = logService.read(subTaskJobLogPath, 0);
            return read;
        } catch (IOException e) {
            log.error(e.getMessage());
            throw new BusinessException("读取日志失败" + e.toString());
        }
    }

    @Override
    public Boolean del(Long docCapId) {

        this.removeById(docCapId);
        return true;
    }

    @Override
    public String createSql(String tableName, List<String> fieldName, Long projectId) {

        // 生成建表语句
        final BaseDbManager dbManager = projectService.getDbManager(projectId);
        final StorageEntity storageEntity = storageServiceImpl.getStorageInfoByProjectId(projectId);
        final DSType dsType = DSType.from(storageEntity.getStorageType());
        final String dbName = storageEntity.getRmdbs().getDbName();
        final String schema = storageEntity.getRmdbs().getSchema();

        if (storageEntity.getStorageType().equals(DSType.Sap.name())
                || storageEntity.getStorageType().equals(DSType.PostgreSQL.name())
                || storageEntity.getStorageType().equalsIgnoreCase(DSType.KINGBASE8.name())) {
            // sap传入schema
            String sql = tableManager.getCreateTableSql(dbManager, fieldName, tableName, schema);
            return sql;
        } else if (storageEntity.getStorageType().equals(DSType.DM.name())) {
            String sql = tableManager.getCreateTableSql(dbManager, fieldName, tableName, dbName);
            return sql;
        } else {
            String sql = tableManager.getCreateTableSql(dbManager, fieldName, tableName, "");
            return sql;
        }
    }

    @Override
    public Boolean updateDocCap(DocumentCaptureEntity documentCaptureEntity) {
        return this.saveOrUpdate(documentCaptureEntity);
    }

    @Override
    public List<DocumentCaptureDomain> getDocTask(Long projectId, String tableName) {

        List<DocumentCaptureEntity> multi =
                this.list(
                        new LambdaQueryWrapper<DocumentCaptureEntity>()
                                .eq(DocumentCaptureEntity::getProjectId, projectId)
                                .eq(DocumentCaptureEntity::getTargetTblName, tableName));

        List<DocumentCaptureDomain> documentCaptureDomains = documentCaptureDomainMapper.map(multi);

        return documentCaptureDomains;
    }

    @Override
    public List<List<String>> parseFile(MultipartFile file, Map<String, String[]> parameterMap) {
        List<List<String>> rows = new ArrayList<>();
        try {
            String fileUrl = file.getOriginalFilename();
            InputStream is = file.getInputStream();

            // 根据文件格式选择不同的解析方法
            if (fileUrl.endsWith("xlsx")) {
                // Excel 2007
                rows = readExcel(is);
            } else if (fileUrl.endsWith("xls")) {
                // Excel 2003
                rows = parseExcel2003(is);
            } else if (fileUrl.endsWith("csv")) {
                // CSV
                String encodeFormat = parameterMap.get("encodeFormat")[0];
                rows = parseCSV(file, encodeFormat);
            } else {
                throw new BusinessException("不支持的文件格式");
            }
        } catch (IOException e) {
            log.error("文件采集IO错误：" + e.toString());
        }
        return rows;
    }

    private String getCellValue(Cell cell) {
        if (cell == null) {
            return "";
        }
        switch (cell.getCellType()) {
            case STRING:
                return cell.getStringCellValue();
            case NUMERIC:
                if (DateUtil.isCellDateFormatted(cell)) {
                    return cell.getDateCellValue().toString();
                } else {
                    return String.valueOf(cell.getNumericCellValue());
                }
            case BOOLEAN:
                return String.valueOf(cell.getBooleanCellValue());
            case FORMULA:
                return cell.getCellFormula();
            default:
                return "";
        }
    }

    public List<List<String>> readExcel(InputStream is) {
        List<List<String>> excelData = new ArrayList<>();
        XMLReader xmlReader = null;
        XSSFHandler handler = null;
        OPCPackage opcPackage = null;
        // 创建XSSFReader
        Iterator<InputStream> sheets = null;
        IOUtils.setByteArrayMaxOverride(1000000000);
        try {
            opcPackage = OPCPackage.open(is);
            XSSFReader xssfReader = new XSSFReader(opcPackage);
            // 获取SharedStringsTable和StylesTable
            SharedStrings sharedStringsTable = xssfReader.getSharedStringsTable();
            StylesTable stylesTable = xssfReader.getStylesTable();

            // 创建XMLReader
            xmlReader = XMLReaderFactory.createXMLReader();
            handler = new XSSFHandler(sharedStringsTable, stylesTable);
            // 设置ContentHandler
            xmlReader.setContentHandler(handler);
            // 获取SheetIterator
            sheets = xssfReader.getSheetsData();
            while (sheets.hasNext()) {
                InputStream sheetInputStream = sheets.next();
                InputSource sheetSource = new InputSource(sheetInputStream);
                try {
                    xmlReader.parse(sheetSource);
                    sheetInputStream.close();
                } catch (Exception e) {

                }
                excelData.addAll(handler.getData());
                opcPackage.close();
            }
        } catch (Exception e) {
            log.error(e.getMessage());
            throw new BusinessException("解析excel失败");
        }
        // 关闭OPCPackage
        return excelData;
    }

    @SneakyThrows
    private void processSheet(Sheet sheet,
                              XSSFSheetXMLHandler.SheetContentsHandler sheetContentsHandler) {
        try {
            XMLReader sheetParser = XMLReaderFactory.createXMLReader();
            XSSFSheetXMLHandler handler = new XSSFSheetXMLHandler(null, null, sheetContentsHandler,
                    false);
            sheetParser.setContentHandler(handler);
            sheetParser.parse(new InputSource(sheet.getSheetName()));
        } catch (SAXException | IOException e) {
            throw new RuntimeException("Error processing sheet", e);
        }
    }

    private List<List<String>> parseExcel2003(InputStream is) throws IOException {
        List<List<String>> rows = new ArrayList<>();
        try (Workbook workbook = new HSSFWorkbook(is)) {
            // 读取Excel内容并返回前10行
            // 示例代码，你需要根据实际情况修改
            for (int i = 0; i < workbook.getNumberOfSheets(); i++) {
                Sheet sheet = workbook.getSheetAt(i);
                for (Row row : sheet) {
                    List<String> rowData = new ArrayList<>();
                    for (Cell cell : row) {
                        String string = cell.toString();
                        rowData.add(string);
                    }
                    rows.add(rowData);
                    if (rows.size() >= 10) {
                        break;
                    }
                }
                if (rows.size() >= 10) {
                    break;
                }
            }
        }
        return rows;
    }

    private List<List<String>> parseCSV(MultipartFile file, String encodeFormat) throws IOException {
        List<List<String>> rows = new ArrayList<>();
        try (BufferedReader reader = new BufferedReader(
                new InputStreamReader(file.getInputStream(), encodeFormat))) {
            String line;
            int rowCount = 0;
            while ((line = reader.readLine()) != null && rowCount < 10) {
                String[] fields = line.split(",");
                List<String> row = Arrays.asList(fields);
                rows.add(row);
                rowCount++;
            }
        }
        return rows;
    }

    @Override
    public void exportFile(Long id, HttpServletResponse response) {
        DocumentCaptureEntity documentCaptureEntity = this.getById(id);
        if (documentCaptureEntity == null) {
            throw new BusinessException("未找到对应的文件记录");
        }

        // 获取文件路径
        String filePath = documentCaptureEntity.getOrgFilePath();
        if (StringUtils.isBlank(filePath)) {
            throw new BusinessException("文件路径为空");
        }
        filePath = filePath.replace("\\", "/");

        // 设置响应头
        String fileName = documentCaptureEntity.getOrgFileName();
        String fileExtension = fileName.substring(fileName.lastIndexOf(".") + 1).toLowerCase();
        switch (fileExtension) {
            case "csv":
                response.setContentType("text/csv");
                break;
            case "txt":
                response.setContentType("text/plain");
                break;
            case "xls":
                response.setContentType("application/vnd.ms-excel");
                break;
            case "xlsx":
                response.setContentType("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet");
                break;
            default:
                throw new BusinessException("不支持的文件格式");
        }
        response.setCharacterEncoding("UTF-8");
        response.setHeader("Content-disposition",
                "attachment; filename=\"" + fileName + "\"");


        // 读取文件并写入响应
        try {
            FileDownloadUtil.download(response, filePath.split("\\?")[1].split("=")[1]);
        } catch (IOException e) {
            log.error("文件导出失败", e);
            throw new BusinessException("文件导出失败: " + e.getMessage());
        }
    }

}
